

/*-------------------------------------------------------------------------
 *
 * commit_ts.c
 *		PostgreSQL 提交时间戳管理器
 *
 * 本模块是一个类似 pg_xact 的系统，用于存储每个事务的提交时间戳。
 *
 * XLOG 交互：只要初始化一个新的 CommitTs 页面为零，本模块就会生成一个 XLOG 记录。
 * 此外，当调用方请求时，为设置值生成一个 XLOG 记录；这使我们能够支持来自
 * 事务提交以外其他地方的值。其他的 CommitTS 写入来自于在 xact.c 
 * 中记录事务提交，这会为这些事件生成自己的 XLOG 记录，并在重做时重新执行状态更新；
 * 所以我们在这里不需要额外的 XLOG 条目。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * src/backend/access/transam/commit_ts.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/commit_ts.h"
#include "access/htup_details.h"
#include "access/slru.h"
#include "access/transam.h"
#include "access/xloginsert.h"
#include "access/xlogutils.h"
#include "catalog/pg_type.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "pg_trace.h"
#include "storage/shmem.h"
#include "utils/builtins.h"
#include "utils/snapmgr.h"
#include "utils/timestamp.h"

/*
 * 定义 CommitTs 页的大小。一个页面的大小与 Postgres 中使用的 BLCKSZ 相同。
 *
 * 注意：由于 TransactionIds 为 32 位并在 0xFFFFFFFF 处回绕，
 * CommitTs 页编号也在 0xFFFFFFFF/COMMIT_TS_XACTS_PER_PAGE 处回绕，
 * CommitTs 段编号在 0xFFFFFFFF/COMMIT_TS_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT 处回绕。
 * 在本模块中，我们无需对此事实给予明确关注，除非在 TruncateCommitTs 中比较段
 * 和页面编号（见 CommitTsPagePrecedes）。
 */

/*
 * 每个 xact 需要 8 + 2 字节。请注意，扩大这个结构可能意味着
 * 最大的文件名长度超过 5 个字符；见 SlruScanDirectory。
 */
typedef struct CommitTimestampEntry
{
	TimestampTz time;
	RepOriginId nodeid;
} CommitTimestampEntry;

#define SizeOfCommitTimestampEntry (offsetof(CommitTimestampEntry, nodeid) + \
									sizeof(RepOriginId))

#define COMMIT_TS_XACTS_PER_PAGE \
	(BLCKSZ / SizeOfCommitTimestampEntry)

#define TransactionIdToCTsPage(xid) \
	((xid) / (TransactionId) COMMIT_TS_XACTS_PER_PAGE)
#define TransactionIdToCTsEntry(xid)	\
	((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE)

/*
 * 链接到 CommitTs 控制的共享内存数据结构
 */
static SlruCtlData CommitTsCtlData;

#define CommitTsCtl (&CommitTsCtlData)

/*
 * 我们保持一个缓存，以保存共享内存中设置的最后一个值。
 *
 * 这也是保持激活状态的好地方。我们将其与 GUC 分开，
 * 以便在主机上独立于 GUC 的值激活模块时，备份可以激活该模块。
 *
 * 这由 CommitTsLock 保护。在某些地方，我们在没有获取锁的情况下使用 commitTsActive；
 * 在发生这种情况时，注释解释了其理由。
 */
typedef struct CommitTimestampShared
{
	TransactionId xidLastCommit;
	CommitTimestampEntry dataLastCommit;
	bool		commitTsActive;
} CommitTimestampShared;

static CommitTimestampShared *commitTsShared;


/* GUC 变量 */
bool		track_commit_timestamp;

static void fc_SetXidCommitTsInPage(TransactionId fc_xid, int fc_nsubxids,
								 TransactionId *fc_subxids, TimestampTz fc_ts,
								 RepOriginId fc_nodeid, int fc_pageno);
static void fc_TransactionIdSetCommitTs(TransactionId fc_xid, TimestampTz fc_ts,
									 RepOriginId fc_nodeid, int fc_slotno);
static void fc_error_commit_ts_disabled(void);
static int	fc_ZeroCommitTsPage(int fc_pageno, bool fc_writeXlog);
static bool fc_CommitTsPagePrecedes(int fc_page1, int fc_page2);
static void fc_ActivateCommitTs(void);
static void fc_DeactivateCommitTs(void);
static void fc_WriteZeroPageXlogRec(int fc_pageno);
static void fc_WriteTruncateXlogRec(int fc_pageno, TransactionId fc_oldestXid);

/*
 * TransactionTreeSetCommitTsData
 *
 * 尽可能高效地记录提交日志中事务及其子事务树的最终提交时间戳。
 *
 * xid 是顶层事务 id。
 *
 * subxids 是一个长度为 nsubxids 的 xid 数组，表示 xid 树中的子事务。
 * 在各种情况下，nsubxids 可能为零。
 * 仅跟踪父 xid 提交时间戳不足的原因是子事务 SLRU
 * 不会在崩溃时保持有效（它不是永久的），因此我们需要在这里
 * 保留关于它们的信息。如果未来子事务的实现发生变化，
 * 我们可能需要重新考虑是否为每个 subxid 存储时间戳信息的决定。
 */
void TransactionTreeSetCommitTsData(TransactionId fc_xid, int fc_nsubxids,
							   TransactionId *fc_subxids, TimestampTz fc_timestamp,
							   RepOriginId fc_nodeid)
{
	int			fc_i;
	TransactionId fc_headxid;
	TransactionId fc_newestXact;

	/*
	 * 如果模块未激活，则无操作。
	 *
	 * 此处无锁读取是可以的，因为在备份中（唯一可以
	 * 更改标志的地方），此例程仅由恢复过程调用，
	 * 该进程也是唯一可以更改标志的进程。
	 */
	if (!commitTsShared->commitTsActive)
		return;

	/*
	 * 找出这一批次中最新的 Xid：如果有任何subxid，则是最后的 subxid，否则是父 xid。
	 */
	if (fc_nsubxids > 0)
		fc_newestXact = fc_subxids[fc_nsubxids - 1];
	else
		fc_newestXact = fc_xid;

	/*
	 * 我们将 xids 拆分为属于同一 SLRU 页的组；每组中的第一个元素是其头部。
	 * 第一组以主 XID 作为头部；后续组使用不在前一页上的第一个 subxid 作为头部。
	 * 这样，我们只需锁定/修改每个 SLRU 页一次。
	 */
	fc_headxid = fc_xid;
	fc_i = 0;
	for (;;)
	{
		int			fc_pageno = TransactionIdToCTsPage(fc_headxid);
		int			fc_j;

		for (fc_j = fc_i; fc_j < fc_nsubxids; fc_j++)
		{
			if (TransactionIdToCTsPage(fc_subxids[fc_j]) != fc_pageno)
				break;
		}
		/* subxids[i..j] 与头部处于同一页 */

		fc_SetXidCommitTsInPage(fc_headxid, fc_j - fc_i, fc_subxids + fc_i, fc_timestamp, fc_nodeid,
							 fc_pageno);

		/* 如果我们写出了所有 subxids，我们就完成了。 */
		if (fc_j >= fc_nsubxids)
			break;

		/*
		 * 设置新的头部并跳过它，以及我们刚刚写的 subxids。
		 */
		fc_headxid = fc_subxids[fc_j];
		fc_i = fc_j + 1;
	}

	/* 在共享内存中更新缓存值 */
	LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
	commitTsShared->xidLastCommit = fc_xid;
	commitTsShared->dataLastCommit.time = fc_timestamp;
	commitTsShared->dataLastCommit.nodeid = fc_nodeid;

	/* 如果需要，则向前移动我们的端点 */
	if (TransactionIdPrecedes(ShmemVariableCache->newestCommitTsXid, fc_newestXact))
		ShmemVariableCache->newestCommitTsXid = fc_newestXact;
	LWLockRelease(CommitTsLock);
}

/*
 * 在提交日志中记录单个页面上所有事务条目的提交时间戳。仅在此页面上是原子性的。
 */
static void fc_SetXidCommitTsInPage(TransactionId fc_xid, int fc_nsubxids,
					 TransactionId *fc_subxids, TimestampTz fc_ts,
					 RepOriginId fc_nodeid, int fc_pageno)
{
	int			fc_slotno;
	int			fc_i;

	LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);

	fc_slotno = SimpleLruReadPage(CommitTsCtl, fc_pageno, true, fc_xid);

	fc_TransactionIdSetCommitTs(fc_xid, fc_ts, fc_nodeid, fc_slotno);
	for (fc_i = 0; fc_i < fc_nsubxids; fc_i++)
		fc_TransactionIdSetCommitTs(fc_subxids[fc_i], fc_ts, fc_nodeid, fc_slotno);

	CommitTsCtl->shared->page_dirty[fc_slotno] = true;

	LWLockRelease(CommitTsSLRULock);
}

/*
 * 设置单个事务的提交时间戳。
 *
 * 必须在持有 CommitTsSLRULock 时调用
 */
static void fc_TransactionIdSetCommitTs(TransactionId fc_xid, TimestampTz fc_ts,
						 RepOriginId fc_nodeid, int fc_slotno)
{
	int			fc_entryno = TransactionIdToCTsEntry(fc_xid);
	CommitTimestampEntry fc_entry;

	Assert(TransactionIdIsNormal(fc_xid));

	fc_entry.time = fc_ts;
	fc_entry.nodeid = fc_nodeid;

	memcpy(CommitTsCtl->shared->page_buffer[fc_slotno] +
		   SizeOfCommitTimestampEntry * fc_entryno,
		   &fc_entry, SizeOfCommitTimestampEntry);
}

/*
 * 查询事务的提交时间戳。
 *
 * 返回值指示是否为给定的 xid 找到了提交时间戳记录。时间戳值返回在 *ts 中（可能不为 null），如果不为 null，则 Xid 的原节点返回在 *nodeid 中。
 */
bool TransactionIdGetCommitTsData(TransactionId fc_xid, TimestampTz *fc_ts,
							 RepOriginId *fc_nodeid)
{
	int			fc_pageno = TransactionIdToCTsPage(fc_xid);
	int			fc_entryno = TransactionIdToCTsEntry(fc_xid);
	int			fc_slotno;
	CommitTimestampEntry fc_entry;
	TransactionId fc_oldestCommitTsXid;
	TransactionId fc_newestCommitTsXid;

	if (!TransactionIdIsValid(fc_xid))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("cannot retrieve commit timestamp for transaction %u", fc_xid)));
	else if (!TransactionIdIsNormal(fc_xid))
	{
		/* 冻结和引导的 xid 始终在过去很久之前提交 */
		*fc_ts = 0;
		if (fc_nodeid)
			*fc_nodeid = 0;
		return false;
	}

	LWLockAcquire(CommitTsLock, LW_SHARED);

	/* 如果模块未启用则出错 */
	if (!commitTsShared->commitTsActive)
		fc_error_commit_ts_disabled();

	/*
	 * 如果请求了缓存值，则返回该值。否则，回退到从 SLRU 读取。
	 */
	if (commitTsShared->xidLastCommit == fc_xid)
	{
		*fc_ts = commitTsShared->dataLastCommit.time;
		if (fc_nodeid)
			*fc_nodeid = commitTsShared->dataLastCommit.nodeid;

		LWLockRelease(CommitTsLock);
		return *fc_ts != 0;
	}

	fc_oldestCommitTsXid = ShmemVariableCache->oldestCommitTsXid;
	fc_newestCommitTsXid = ShmemVariableCache->newestCommitTsXid;
	/* 既无效，或两者都无效 */
	Assert(TransactionIdIsValid(fc_oldestCommitTsXid) == TransactionIdIsValid(fc_newestCommitTsXid));
	LWLockRelease(CommitTsLock);

	/*
	 * 如果请求的值超出我们的有效范围，则返回空。
	 */
	if (!TransactionIdIsValid(fc_oldestCommitTsXid) ||
		TransactionIdPrecedes(fc_xid, fc_oldestCommitTsXid) ||
		TransactionIdPrecedes(fc_newestCommitTsXid, fc_xid))
	{
		*fc_ts = 0;
		if (fc_nodeid)
			*fc_nodeid = InvalidRepOriginId;
		return false;
	}

	/* 锁由 SimpleLruReadPage_ReadOnly 获取 */
	fc_slotno = SimpleLruReadPage_ReadOnly(CommitTsCtl, fc_pageno, fc_xid);
	memcpy(&fc_entry,
		   CommitTsCtl->shared->page_buffer[fc_slotno] +
		   SizeOfCommitTimestampEntry * fc_entryno,
		   SizeOfCommitTimestampEntry);

	*fc_ts = fc_entry.time;
	if (fc_nodeid)
		*fc_nodeid = fc_entry.nodeid;

	LWLockRelease(CommitTsSLRULock);
	return *fc_ts != 0;
}

/*
 * 返回最新已提交事务的 Xid。（就这个模块而言；确保值对调用者有用是调用者的责任。）
 *
 * ts 和 nodeid 填充对应的数据；如果不需要，可以传递为 NULL。
 */
TransactionId GetLatestCommitTsData(TimestampTz *fc_ts, RepOriginId *fc_nodeid)
{
	TransactionId fc_xid;

	LWLockAcquire(CommitTsLock, LW_SHARED);

	/* 如果模块未启用则出错 */
	if (!commitTsShared->commitTsActive)
		fc_error_commit_ts_disabled();

	fc_xid = commitTsShared->xidLastCommit;
	if (fc_ts)
		*fc_ts = commitTsShared->dataLastCommit.time;
	if (fc_nodeid)
		*fc_nodeid = commitTsShared->dataLastCommit.nodeid;
	LWLockRelease(CommitTsLock);

	return fc_xid;
}

static void fc_error_commit_ts_disabled(void)
{
	ereport(ERROR,
			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
			 errmsg("could not get commit timestamp data"),
			 RecoveryInProgress() ?
			 errhint("Make sure the configuration parameter \"%s\" is set on the primary server.",
					 "track_commit_timestamp") :
			 errhint("Make sure the configuration parameter \"%s\" is set.",
					 "track_commit_timestamp")));
}

/*
 * 用于获取事务提交时间的 SQL 可调用包装器
 */
Datum pg_xact_commit_timestamp(PG_FUNCTION_ARGS)
{
	TransactionId fc_xid = PG_GETARG_TRANSACTIONID(0);
	TimestampTz fc_ts;
	bool		fc_found;

	fc_found = TransactionIdGetCommitTsData(fc_xid, &fc_ts, NULL);

	if (!fc_found)
		PG_RETURN_NULL();

	PG_RETURN_TIMESTAMPTZ(fc_ts);
}


/*
 * pg_last_committed_xact
 *
 * 用于获取有关最新已提交事务的一些信息的 SQL 可调用包装器：事务 ID、时间戳和复制来源。
 */
Datum pg_last_committed_xact(PG_FUNCTION_ARGS)
{
	TransactionId fc_xid;
	RepOriginId fc_nodeid;
	TimestampTz fc_ts;
	Datum		fc_values[3];
	bool		fc_nulls[3];
	TupleDesc	fc_tupdesc;
	HeapTuple	fc_htup;

	/* 并构造一个包含我们数据的元组 */
	fc_xid = GetLatestCommitTsData(&fc_ts, &fc_nodeid);

	/*
	 * 为结果行构造一个元组描述符。必须与此函数的 pg_proc 条目匹配！
	 */
	fc_tupdesc = CreateTemplateTupleDesc(3);
	TupleDescInitEntry(fc_tupdesc, (AttrNumber) 1, "xid",
					   XIDOID, -1, 0);
	TupleDescInitEntry(fc_tupdesc, (AttrNumber) 2, "timestamp",
					   TIMESTAMPTZOID, -1, 0);
	TupleDescInitEntry(fc_tupdesc, (AttrNumber) 3, "roident",
					   OIDOID, -1, 0);
	fc_tupdesc = BlessTupleDesc(fc_tupdesc);

	if (!TransactionIdIsNormal(fc_xid))
	{
		memset(fc_nulls, true, sizeof(fc_nulls));
	}
	else
	{
		fc_values[0] = TransactionIdGetDatum(fc_xid);
		fc_nulls[0] = false;

		fc_values[1] = TimestampTzGetDatum(fc_ts);
		fc_nulls[1] = false;

		fc_values[2] = ObjectIdGetDatum((Oid) fc_nodeid);
		fc_nulls[2] = false;
	}

	fc_htup = heap_form_tuple(fc_tupdesc, fc_values, fc_nulls);

	PG_RETURN_DATUM(HeapTupleGetDatum(fc_htup));
}

/*
 * pg_xact_commit_timestamp_origin
 *
 * 用于获取给定事务的提交时间戳和复制来源的 SQL 可调用包装器。
 */
Datum pg_xact_commit_timestamp_origin(PG_FUNCTION_ARGS)
{
	TransactionId fc_xid = PG_GETARG_TRANSACTIONID(0);
	RepOriginId fc_nodeid;
	TimestampTz fc_ts;
	Datum		fc_values[2];
	bool		fc_nulls[2];
	TupleDesc	fc_tupdesc;
	HeapTuple	fc_htup;
	bool		fc_found;

	fc_found = TransactionIdGetCommitTsData(fc_xid, &fc_ts, &fc_nodeid);

	/*
	 * 为结果行构造一个元组描述符。必须与此函数的 pg_proc 条目匹配！
	 */
	fc_tupdesc = CreateTemplateTupleDesc(2);
	TupleDescInitEntry(fc_tupdesc, (AttrNumber) 1, "timestamp",
					   TIMESTAMPTZOID, -1, 0);
	TupleDescInitEntry(fc_tupdesc, (AttrNumber) 2, "roident",
					   OIDOID, -1, 0);
	fc_tupdesc = BlessTupleDesc(fc_tupdesc);

	if (!fc_found)
	{
		memset(fc_nulls, true, sizeof(fc_nulls));
	}
	else
	{
		fc_values[0] = TimestampTzGetDatum(fc_ts);
		fc_nulls[0] = false;

		fc_values[1] = ObjectIdGetDatum((Oid) fc_nodeid);
		fc_nulls[1] = false;
	}

	fc_htup = heap_form_tuple(fc_tupdesc, fc_values, fc_nulls);

	PG_RETURN_DATUM(HeapTupleGetDatum(fc_htup));
}

/*
 * 共享 CommitTS 缓冲区的数量。
 *
 * 我们使用与 CLOG 缓冲区数量非常相似的逻辑（只是我们对共享缓冲区的扩展速度是原来的两倍，且最大值是两倍）；请参见 CLOGShmemBuffers 中的注释。
 */
Size CommitTsShmemBuffers(void)
{
	return Min(256, Max(4, NBuffers / 256));
}

/*
 * CommitTs 的共享内存大小设置
 */
Size CommitTsShmemSize(void)
{
	return SimpleLruShmemSize(CommitTsShmemBuffers(), 0) +
		sizeof(CommitTimestampShared);
}

/*
 * 在系统启动时初始化 CommitTs（主进程启动或独立后端）
 */
void CommitTsShmemInit(void)
{
	bool		fc_found;

	CommitTsCtl->PagePrecedes = fc_CommitTsPagePrecedes;
	SimpleLruInit(CommitTsCtl, "CommitTs", CommitTsShmemBuffers(), 0,
				  CommitTsSLRULock, "pg_commit_ts",
				  LWTRANCHE_COMMITTS_BUFFER,
				  SYNC_HANDLER_COMMIT_TS);
	SlruPagePrecedesUnitTests(CommitTsCtl, COMMIT_TS_XACTS_PER_PAGE);

	commitTsShared = ShmemInitStruct("CommitTs shared",
									 sizeof(CommitTimestampShared),
									 &fc_found);

	if (!IsUnderPostmaster)
	{
		Assert(!fc_found);

		commitTsShared->xidLastCommit = InvalidTransactionId;
		TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time);
		commitTsShared->dataLastCommit.nodeid = InvalidRepOriginId;
		commitTsShared->commitTsActive = false;
	}
	else
		Assert(fc_found);
}

/*
 * 此函数必须在系统安装时调用一次。
 *
 * （假设 CommitTs 目录已由 initdb 创建，并且 CommitTsShmemInit 已经被调用。）
 */
void BootStrapCommitTs(void)
{
	/*
	 * 目前这里没有需要做的工作，与其他大多数 SLRU 模块不同；当服务器启动时启用此模块时会创建段。请参见 ActivateCommitTs。
	 */
}

/*
 * 初始化（或重新初始化）CommitTs的一页为零。
 * 如果writeXlog为真，还会发出一个XLOG记录，表明我们这样做了。
 *
 * 页面实际上并没有被写入，只是在共享内存中设置。
 * 新页面的槽号将被返回。
 *
 * 入口时必须持有控制锁，并将在退出时保持。
 */
static int fc_ZeroCommitTsPage(int fc_pageno, bool fc_writeXlog)
{
	int			fc_slotno;

	fc_slotno = SimpleLruZeroPage(CommitTsCtl, fc_pageno);

	if (fc_writeXlog)
		fc_WriteZeroPageXlogRec(fc_pageno);

	return fc_slotno;
}

/*
 * 在 postmaster 或独立后端启动期间，必须在 StartupXLOG 初始化
 * ShmemVariableCache->nextXid 后调用一次。
 */
void StartupCommitTs(void)
{
	fc_ActivateCommitTs();
}

/*
 * 这个必须在postmaster或独立后端启动期间调用一次，
 * 在恢复完成之后。
 */
void CompleteCommitTsInitialization(void)
{
	/*
	 * 如果该功能未启用，则彻底关闭它。这也会删除
	 * 任何残留数据。
	 *
	 * 另一方面，如果该功能启用，则激活该模块。这
	 * 对于主节点和备用节点是必要的，因为激活依赖于
	 * 恢复开始时或重放XLOG_PARAMETER_CHANGE时
	 * 控制文件的内容。
	 */
	if (!track_commit_timestamp)
		fc_DeactivateCommitTs();
	else
		fc_ActivateCommitTs();
}

/*
 * 在恢复期间接收到XLOG_PARAMETER_CHANGE
 * XLog记录时激活或停用CommitTs。
 */
void CommitTsParameterChange(bool fc_newvalue, bool fc_oldvalue)
{
	/*
	 * 如果在此服务器中commit_ts模块被禁用，而我们从
	 * 主服务器那里得知它在那边被启用，则激活它，以便我们
	 * 可以重放涉及它的未来WAL记录；同时在
	 * pg_control上标记为活动。如果旧值已经设置，
	 * 我们已完成此操作，因此无需再做任何事情。
	 *
	 * 如果主节点中模块被禁用，那么这里也要禁用它，除非
	 * 模块在本地被启用。
	 *
	 * 请注意，这仅在恢复过程中运行，因此无锁读取是
	 * 可以的。
	 */
	if (fc_newvalue)
	{
		if (!commitTsShared->commitTsActive)
			fc_ActivateCommitTs();
	}
	else if (commitTsShared->commitTsActive)
		fc_DeactivateCommitTs();
}

/*
 * 在必要时激活该模块。
 *		这必须在postmaster或独立后端启动期间发生，
 *		或在主节点中track_commit_timestamp设置
 *		更改的任何时候进行WAL重放时。
 *
 * 该SLRU需要单独的激活/停用功能的原因是，
 * 它可以在启动期间启用/禁用，并且主节点的激活/停用
 * 会通过重放传播到备用节点。其他SLRU没有这个特性
 * ，它们可以在正常启动过程中初始化。
 *
 * 这负责创建当前活动的段，如果它尚不存在的话。
 * 这样做的原因是服务器可能已经以禁用该模块的模式
 * 运行了一段时间，因此可能会跳过正常的创建点。
 */
static void fc_ActivateCommitTs(void)
{
	TransactionId fc_xid;
	int			fc_pageno;

	/* 如果我们已经完成这个，就没有什么可做的 */
	LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
	if (commitTsShared->commitTsActive)
	{
		LWLockRelease(CommitTsLock);
		return;
	}
	LWLockRelease(CommitTsLock);

	fc_xid = XidFromFullTransactionId(ShmemVariableCache->nextXid);
	fc_pageno = TransactionIdToCTsPage(fc_xid);

	/*
	 * 重新初始化我们对最新页面编号的概念。
	 */
	LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
	CommitTsCtl->shared->latest_page_number = fc_pageno;
	LWLockRelease(CommitTsSLRULock);

	/*
	 * 如果启用了 CommitTs，但在之前的服务器运行中没有启用，我们
	 * 需要将最旧和最新的值设置为下一个 Xid；这样我们就不会尝试读取
	 * 可能未设置的数据。
	 *
	 * XXX 如果服务器启动时启用了 commitTs，然后启动时禁用了 commitTs，
	 * 再次启动时启用它，这会有问题吗？看起来不会，因为在恢复结束时
	 * 应该有一个检查点将值设置为 InvalidTransactionId；因此，在
	 * oldestCommitTsXid 被暂时设置为 Invalid 之后，注入没有 CommitTs
	 * 值的新事务的任何机会都不会发生。
	 */
	LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
	if (ShmemVariableCache->oldestCommitTsXid == InvalidTransactionId)
	{
		ShmemVariableCache->oldestCommitTsXid =
			ShmemVariableCache->newestCommitTsXid = ReadNextTransactionId();
	}
	LWLockRelease(CommitTsLock);

	/* 如有必要，创建当前的段文件 */
	if (!SimpleLruDoesPhysicalPageExist(CommitTsCtl, fc_pageno))
	{
		int			fc_slotno;

		LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
		fc_slotno = fc_ZeroCommitTsPage(fc_pageno, false);
		SimpleLruWritePage(CommitTsCtl, fc_slotno);
		Assert(!CommitTsCtl->shared->page_dirty[fc_slotno]);
		LWLockRelease(CommitTsSLRULock);
	}

	/* 在共享内存中改变激活状态。 */
	LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
	commitTsShared->commitTsActive = true;
	LWLockRelease(CommitTsLock);
}

/*
 * 禁用此模块。
 *
 * 当 track_commit_timestamp 参数被关闭时必须调用此函数。
 * 这发生在主进程或独立后端启动期间，或在 WAL 重放期间。
 *
 * 重置 CommitTs 进入无效状态，以确保我们不会返回
 * 可能无效的数据；还删除旧数据的段。
 */
static void fc_DeactivateCommitTs(void)
{
	/*
	 * 清理共享内存中的状态。
	 *
	 * 我们重置 commitTsShared 记录中的所有内容，以防止用户在
	 * 模块在主机上重复激活时获取关于备用机上最后提交事务的
	 * 混淆数据。
	 */
	LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);

	commitTsShared->commitTsActive = false;
	commitTsShared->xidLastCommit = InvalidTransactionId;
	TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time);
	commitTsShared->dataLastCommit.nodeid = InvalidRepOriginId;

	ShmemVariableCache->oldestCommitTsXid = InvalidTransactionId;
	ShmemVariableCache->newestCommitTsXid = InvalidTransactionId;

	LWLockRelease(CommitTsLock);

	/*
	 * 删除 *所有* 文件。这是必要的，以确保没有剩余的
	 * 文件；如果此功能在禁用一段时间后再次启用，文件序列中可能
	 * 存在缺口。
	 * （我们可能可以容忍无序文件，因为它们在我们重新循环时会被覆盖，
	 * 但保持整洁似乎更好。）
	 */
	LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
	(void) SlruScanDirectory(CommitTsCtl, SlruScanDirCbDeleteAll, NULL);
	LWLockRelease(CommitTsSLRULock);
}

/*
 * 执行检查点 --- 无论是在关闭期间还是动态执行
 */
void CheckPointCommitTs(void)
{
	/*
	 * 将脏的 CommitTs 页面写入磁盘。这可能导致同步请求
	 * 排队，稍后由 ProcessSyncRequests() 处理，作为
	 * 检查点的一部分。
	 */
	SimpleLruWriteAll(CommitTsCtl, true);
}

/*
 * 确保 CommitTs 有空间为新分配的 XID。
 *
 * 注意：在持有 XidGenLock 时调用此函数。我们希望大部分时间
 * 它能够非常快速；即使在不那么快速的时候，除非我们被迫写出
 * 脏的 CommitTs 或 xlog 页面以在共享内存中腾出空间，否则不会
 * 发生实际的 I/O。
 *
 * 注意：当前实现依赖于 track_commit_timestamp 为
 * PGC_POSTMASTER。
 */
void ExtendCommitTs(TransactionId fc_newestXact)
{
	int			fc_pageno;

	/*
	 * 如果模块未启用则无需执行任何操作。请注意，我们在此处
	 * 进行无锁的标志读取，这是可以的，因为此例程仅由
	 * GetNewTransactionId 调用，而该调用在备用机上
	 * 从未发生。
	 */
	Assert(!InRecovery);
	if (!commitTsShared->commitTsActive)
		return;

	/*
	 * 除了页面的第一个 XID，没什么工作要做。但要注意：
	 * 刚刚环绕过后，页面零的第一个 XID 是
	 * FirstNormalTransactionId。
	 */
	if (TransactionIdToCTsEntry(fc_newestXact) != 0 &&
		!TransactionIdEquals(fc_newestXact, FirstNormalTransactionId))
		return;

	fc_pageno = TransactionIdToCTsPage(fc_newestXact);

	LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);

	/* 将页面置零并创建一个关于它的 XLOG 条目 */
	fc_ZeroCommitTsPage(fc_pageno, !InRecovery);

	LWLockRelease(CommitTsSLRULock);
}

/*
 * 删除在传入事务 ID 之前的所有 CommitTs 段落。
 *
 * 注意，我们在这里无需刷新 XLOG。
 */
void TruncateCommitTs(TransactionId fc_oldestXact)
{
	int			fc_cutoffPage;

	/*
	 * 截止点是包含 oldestXact 的段的开始。我们
	 * 将包含 oldestXact 的 *页面* 传递给 SimpleLruTruncate。
	 */
	fc_cutoffPage = TransactionIdToCTsPage(fc_oldestXact);

	/* 检查是否有任何可以删除的文件 */
	if (!SlruScanDirectory(CommitTsCtl, SlruScanDirCbReportPresence,
						   &fc_cutoffPage))
		return;					/* 没有什么可删除的 */

	/* 写入 XLOG 记录 */
	fc_WriteTruncateXlogRec(fc_cutoffPage, fc_oldestXact);

	/* 现在我们可以删除旧的 CommitTs 段落 */
	SimpleLruTruncate(CommitTsCtl, fc_cutoffPage);
}

/*
 * 设置可以查询 commit TS 的限制值。
 */
void SetCommitTsLimit(TransactionId fc_oldestXact, TransactionId fc_newestXact)
{
	/*
	 * 小心不要覆盖那些时间上“更未来”或者标志禁用
	 * committs 的值。
	 */
	LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
	if (ShmemVariableCache->oldestCommitTsXid != InvalidTransactionId)
	{
		if (TransactionIdPrecedes(ShmemVariableCache->oldestCommitTsXid, fc_oldestXact))
			ShmemVariableCache->oldestCommitTsXid = fc_oldestXact;
		if (TransactionIdPrecedes(fc_newestXact, ShmemVariableCache->newestCommitTsXid))
			ShmemVariableCache->newestCommitTsXid = fc_newestXact;
	}
	else
	{
		Assert(ShmemVariableCache->newestCommitTsXid == InvalidTransactionId);
		ShmemVariableCache->oldestCommitTsXid = fc_oldestXact;
		ShmemVariableCache->newestCommitTsXid = fc_newestXact;
	}
	LWLockRelease(CommitTsLock);
}

/*
 * 向前推进可以查询的最旧 commitTS 值
 */
void AdvanceOldestCommitTsXid(TransactionId fc_oldestXact)
{
	LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
	if (ShmemVariableCache->oldestCommitTsXid != InvalidTransactionId &&
		TransactionIdPrecedes(ShmemVariableCache->oldestCommitTsXid, fc_oldestXact))
		ShmemVariableCache->oldestCommitTsXid = fc_oldestXact;
	LWLockRelease(CommitTsLock);
}


/*
 * 决定 commitTS 页码在截断目的上是否“更旧”。
 * 类似于 CLOGPagePrecedes()。
 *
 * 在默认 BLCKSZ 下，(1 << 31) % COMMIT_TS_XACTS_PER_PAGE == 128。这
 * 在与 CLOG 和其他 SLRUs 的 (1 << 31) % per_page == 0 的情况下引入了差异。
 * 这个函数从不精确检查
 * TransactionIdPrecedes(x-2^31, x)。当系统达到 xidStopLimit 时，
 * oldestXact 和最新分配的 XID 之间可能存在两个页面边界计数，
 * 具体取决于 oldestXact 是否在其页的前 128 项内。
 * 由于该函数不知道 oldestXact 在 page2 中的位置，
 * 它会对一个实际上可被删除的页面返回 false。这是一个更广泛（但仍可忽略不计）的
 * 截断机会，CLOGPagePrecedes() 无法识别。
 *
 * 作为一个完整的示例，数值项的十进制值使得 page1==1 的条目范围从
 * 1.0 到 1.999。让 N+0.15 是 2^31 条目将跨越的页面数量（N 是一个整数）。
 * 如果 oldestXact=N+2.1，则最终安全的 XID 分配会使得 newestXact=1.95。
 * 我们保留第 2 页，因为条目=2.85 是切换条目是否优先于 oldestXact 页的最后一项的边界。
 * 当 oldestXact=N+2.1 时，第 2 页是可以删除的，而在 oldestXact=N+2.9 时则是宝贵的。
 */
static bool fc_CommitTsPagePrecedes(int fc_page1, int fc_page2)
{
	TransactionId fc_xid1;
	TransactionId fc_xid2;

	fc_xid1 = ((TransactionId) fc_page1) * COMMIT_TS_XACTS_PER_PAGE;
	fc_xid1 += FirstNormalTransactionId + 1;
	fc_xid2 = ((TransactionId) fc_page2) * COMMIT_TS_XACTS_PER_PAGE;
	fc_xid2 += FirstNormalTransactionId + 1;

	return (TransactionIdPrecedes(fc_xid1, fc_xid2) &&
			TransactionIdPrecedes(fc_xid1, fc_xid2 + COMMIT_TS_XACTS_PER_PAGE - 1));
}


/*
 * 写入 ZEROPAGE xlog 记录
 */
static void fc_WriteZeroPageXlogRec(int fc_pageno)
{
	XLogBeginInsert();
	XLogRegisterData((char *) (&fc_pageno), sizeof(int));
	(void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_ZEROPAGE);
}

/*
 * 写入 TRUNCATE xlog 记录
 */
static void fc_WriteTruncateXlogRec(int fc_pageno, TransactionId fc_oldestXid)
{
	xl_commit_ts_truncate fc_xlrec;

	fc_xlrec.pageno = fc_pageno;
	fc_xlrec.oldestXid = fc_oldestXid;

	XLogBeginInsert();
	XLogRegisterData((char *) (&fc_xlrec), SizeOfCommitTsTruncate);
	(void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_TRUNCATE);
}

/*
 * CommitTS 资源管理器的例程
 */
void commit_ts_redo(XLogReaderState *fc_record)
{
	uint8		fc_info = XLogRecGetInfo(fc_record) & ~XLR_INFO_MASK;

	/* 在 commit_ts 记录中不使用备用块 */
	Assert(!XLogRecHasAnyBlockRefs(fc_record));

	if (fc_info == COMMIT_TS_ZEROPAGE)
	{
		int			fc_pageno;
		int			fc_slotno;

		memcpy(&fc_pageno, XLogRecGetData(fc_record), sizeof(int));

		LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);

		fc_slotno = fc_ZeroCommitTsPage(fc_pageno, false);
		SimpleLruWritePage(CommitTsCtl, fc_slotno);
		Assert(!CommitTsCtl->shared->page_dirty[fc_slotno]);

		LWLockRelease(CommitTsSLRULock);
	}
	else if (fc_info == COMMIT_TS_TRUNCATE)
	{
		xl_commit_ts_truncate *fc_trunc = (xl_commit_ts_truncate *) XLogRecGetData(fc_record);

		AdvanceOldestCommitTsXid(fc_trunc->oldestXid);

		/*
		 * 在 XLOG 重放期间，latest_page_number 尚未设置；
		 * 插入一个合适的值以绕过 SimpleLruTruncate 中的合理性测试。
		 */
		CommitTsCtl->shared->latest_page_number = fc_trunc->pageno;

		SimpleLruTruncate(CommitTsCtl, fc_trunc->pageno);
	}
	else
		elog(PANIC, "commit_ts_redo: unknown op code %u", fc_info);
}

/*
 * sync.c 的入口点用于同步 commit_ts 文件。
 */
int committssyncfiletag(const FileTag *fc_ftag, char *fc_path)
{
	return SlruSyncFileTag(CommitTsCtl, fc_ftag, fc_path);
}
